﻿#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "TextProtocol.h"
#include "UpwardProtocol.h"
#include "DownwardProtocol.h"
#include "TextUpwardProtocol.h"
#include <string>
#include <hadoop/SerialUtils.hh>
#include <string.h> //strchr

namespace HadoopPipes {
	TextProtocol::TextProtocol(FILE* down, DownwardProtocol* _handler, FILE* up) {
		downStream = down;
		uplink = new TextUpwardProtocol(up);
		handler = _handler;
	}

	int TextProtocol::readUpto(std::string& buffer, const char* limit) {
		int ch;
		buffer.clear();
		while ((ch = getc(downStream)) != -1) {
			if (strchr(limit, ch) != NULL) {
				return ch;
			}
			buffer += ch;
		}
		return -1;
	}

	void TextProtocol::nextEvent() {
		std::string command;
		std::string arg;
		int sep;
		sep = readUpto(command, delim);
		if (command == "mapItem") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(key, delim);
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(value, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->mapItem(key, value);
		} else if (command == "reduceValue") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(value, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->reduceValue(value);
		} else if (command == "reduceKey") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(key, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->reduceKey(key);
		} else if (command == "start") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(arg, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->start(HadoopUtils::toInt(arg));
		} else if (command == "setJobConf") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(arg, delim);
			int len = HadoopUtils::toInt(arg);
			std::vector<std::string> values(len);
			for(int i=0; i < len; ++i) {
				HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
				sep = readUpto(arg, delim);
				values.push_back(arg);
			}
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->setJobConf(values);
		} else if (command == "setInputTypes") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(key, delim);
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(value, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->setInputTypes(key, value);
		} else if (command == "runMap") {
			std::string split;
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(split, delim);
			std::string reduces;
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(reduces, delim);
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(arg, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->runMap(split, HadoopUtils::toInt(reduces), HadoopUtils::toBool(arg));
		} else if (command == "runReduce") {
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			sep = readUpto(arg, delim);
			HADOOP_ASSERT(sep == '\t', "Short text protocol command " + command);
			std::string piped;
			sep = readUpto(piped, delim);
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->runReduce(HadoopUtils::toInt(arg), HadoopUtils::toBool(piped));
		} else if (command == "abort") { 
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->abort();
		} else if (command == "close") {
			HADOOP_ASSERT(sep == '\n', "Long text protocol command " + command);
			handler->close();
		} else {
			throw HadoopUtils::Error("Illegal text protocol command " + command);
		}
	}
	TextProtocol::~TextProtocol() {
		delete uplink;
	}
}
